package com.macaline.technology.rabbitMq;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;

/**
 * rabbitMQ基本工具类
 * @author warm_joy
 */
public class RabbitMqUtil {

	public static String ip = "127.0.0.1"; // IP地址
	public static int port = 5672; // 端口号
	public static String username = "admin"; // 用户名
	public static String password = "admin"; // 密码
	public static String virtualHost = "/admin";

	private static ConnectionFactory factory = null;

	static {
		factory = new ConnectionFactory();
		factory.setHost(ip);
		factory.setPort(port);
		factory.setVirtualHost(virtualHost);
		factory.setUsername(username);
		factory.setPassword(password);
	}

	public static ConnectionFactory getFactory(){
		return factory;
	}

	/**
	 * 获取MQ连接工具
	 * @author warm_joy
	 * @return
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static Connection getConnection() throws IOException, TimeoutException {
		return getFactory().newConnection();
	}
	public static Connection getConnection(String connectionName) throws IOException, TimeoutException {
		return getFactory().newConnection(connectionName);
	}

	/**
	 * 工作队列模式发送消息
	 * @author warm_joy
	 * @param queueName 队列名称
	 * @param message   消息内容
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void sendMessage(String queueName, String message) throws IOException, TimeoutException {
		Connection connection = RabbitMqUtil.getConnection();
		Channel channel = connection.createChannel();
		channel.queueDeclare(queueName, true, false, false, null);
		channel.basicPublish("", queueName, null, message.getBytes());
		channel.close();
		connection.close();
	}

	/**
	 * 发布与订阅模式下发送Mq消息 注：请确保该交换机已通过bindQueueFanout方法进行绑定队列操作
	 *
	 * @author warm_joy
	 * @param exchangeName 进行消息转发的交换机的名称
	 * @param message      发送的消息内容
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void sendMessageFanout(String exchangeName, String message) throws IOException, TimeoutException {
		Connection connection = getConnection();
		Channel channel = connection.createChannel();
		channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT);
		channel.basicPublish(exchangeName, "", null, message.getBytes());
		channel.close();
		connection.close();
	}

	/**
	 * 发布订阅模式下队列与交换机的绑定
	 *
	 * @author warm_joy
	 * @param exchangeName 交换机名称
	 * @param queueNames   绑定到该交换机下的队列的名称
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void bindQueueFanout(String exchangeName, List<String> queueNames)
			throws IOException, TimeoutException {
		Connection connection = getConnection();
		Channel channel = connection.createChannel();
		channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT);
		Iterator<String> iterator = queueNames.iterator();
		while (iterator.hasNext()) {
			String queueName = iterator.next();
			channel.queueDeclare(queueName, true, false, false, null);
			channel.queueBind(queueName, exchangeName, "");
		}
		channel.close();
		connection.close();
	}

	/**
	 * 路由模式下发送Mq消息 注：请确保该交换机已通过sendMessageDirect方法进行路由绑定队列操作
	 *
	 * @author warm_joy
	 * @param exchangeName 进行消息转发的交换机的名称
	 * @param message      发送的消息内容
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void sendMessageDirect(String exchangeName, String routingKey, String message)
			throws IOException, TimeoutException {
		Connection connection = RabbitMqUtil.getConnection();
		Channel channel = connection.createChannel();
		channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
		channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
		channel.close();
		connection.close();
	}

	/**
	 * 路由模式下队列与交换机的绑定
	 *
	 * @author warm_joy
	 * @param exchangeName 交换机的名称
	 * @param queueNames   key：路由key值 value：queue队列名称
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void bindQueueDirect(String exchangeName, Map<String, String> queueNames)
			throws IOException, TimeoutException {
		Connection connection = getConnection();
		Channel channel = connection.createChannel();
		channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
		queueNames.forEach((routingKey, queueName) -> {
			try {
				channel.queueDeclare(queueName, true, false, false, null);
				channel.queueBind(queueName, exchangeName, routingKey);
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		});
		channel.close();
		connection.close();
	}

	/**
	 * 过期队列消息法搜昂
	 * @author warm_joy
	 * @param queueName   队名名称
	 * @param message     消息内容
	 * @param xMessageTtl 消息过期时间
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void sendMessageTLL(String queueName, String message, int xMessageTtl)
			throws IOException, TimeoutException {
		Connection connection = getConnection();
		Channel channel = connection.createChannel();
		Map<String, Object> args = new HashMap<String, Object>();
		args.put("x-message-ttl", xMessageTtl);
		channel.queueDeclare(queueName, true, false, false, args);
		channel.basicPublish("", queueName, null, message.getBytes());
		channel.close();
		connection.close();
	}

	public static void main(String[] args) throws IOException, TimeoutException {

		String queue001 = "queue001";
		String queue002 = "queue002";
		String queue003 = "queue003";
		String queue004 = "queue004";
		String queue005 = "queue005";
		String queue006 = "queue006";


		String exchange1 = "exchange1";
		String exchange2 = "exchange2";
		String exchange3 = "exchange3";



		String routerExchange = "routerExchange";
		String router1 = "router1";
		String router2 = "router2";
		String router3 = "router3";
		LinkedList<String> queueList = new LinkedList<>();
		HashMap<String, String> routerMap = new HashMap<>();
		queueList.add(queue001);
		bindQueueFanout(exchange1, queueList);
		routerMap.put(router1, queue004);
		bindQueueDirect(routerExchange, routerMap);



		queueList.clear();
		queueList.add(queue002);
		bindQueueFanout(exchange2, queueList);
		routerMap.put(router2, queue005);
		bindQueueDirect(routerExchange, routerMap);


		queueList.clear();
		queueList.add(queue001);
		queueList.add(queue002);
		bindQueueFanout(exchange3, queueList);

		routerMap.put(router3, queue006);
		bindQueueDirect(routerExchange, routerMap);

		sendMessageFanout(exchange1, "queue001");
		sendMessageFanout(exchange2, "queue002");
		sendMessageFanout(exchange3, "queue001_002");

		sendMessageDirect(routerExchange, router1, "queue004");
		sendMessageDirect(routerExchange, router2, "queue005");
		sendMessageDirect(routerExchange, router3, "queue006");




	}


}
